package com.sync.data.job.hcp;

import cn.hutool.core.util.NumberUtil;
import com.sync.data.ievent.entity.Hcp;
import com.sync.data.ievent.entity.MtSpeakerSchema;
import com.sync.data.ievent.repository.HcpRepository;
import com.sync.data.ievent.repository.MtSpeakerSchemaRepository;
import com.sync.data.mce.entity.CoreDict;
import com.sync.data.mce.entity.MdmHcp;
import com.sync.data.mce.entity.MdmSpeaker;
import com.sync.data.mce.repository.CoreDictRepository;
import com.sync.data.mce.repository.MdmHcpRepository;
import com.sync.data.mce.repository.MdmSpeakerRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.hibernate.NonUniqueResultException;
import org.hibernate.exception.ConstraintViolationException;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.listener.ExecutionContextPromotionListener;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.data.RepositoryItemReader;
import org.springframework.batch.item.database.JpaItemWriter;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
import org.springframework.core.annotation.Order;
import org.springframework.dao.DataIntegrityViolationException;
import org.springframework.dao.IncorrectResultSizeDataAccessException;
import org.springframework.data.domain.Sort;
import org.springframework.transaction.PlatformTransactionManager;

import java.util.*;
import java.util.stream.Collectors;

@Configuration
@EnableBatchProcessing
@Order(2)
@RequiredArgsConstructor
@Slf4j
public class SpeakerAndHcpBatchJob {

    private final JobBuilderFactory jobBuilderFactory;

    private final StepBuilderFactory stepBuilderFactory;



    private final PlatformTransactionManager transactionManagerIevent;

    private final JpaItemWriter<Hcp> hcpJpaItemWriter;
    private final JpaItemWriter<MtSpeakerSchema> mtSpeakerSchemaWriter;

    // 批处理大小
    private static final int CHUNK_SIZE = 100;
    private final MdmHcpRepository mdmHcpRepository;
    private final MdmSpeakerRepository mdmSpeakerRepository;
    private final HcpRepository hcpRepository;
    private final MtSpeakerSchemaRepository mtSpeakerSchemaRepository;
    private final CoreDictRepository coreDictRepository;

    @Bean
    public Job speakerBatchJob(Step syncHcpStep) {
        return jobBuilderFactory.get("speakerBatchJob")
                .incrementer(new RunIdIncrementer())
                .start(prepareHcpDeptCoreDictStep())
                .next(syncHcpStep)
                .next(syncSpeakerStep())
                .build();
    }


    public Step prepareHcpDeptCoreDictStep() {
        return stepBuilderFactory.get("prepareHcpDeptCoreDictStep")
                .tasklet((contribution, chunkContext) -> {
                    List<CoreDict> bydObjectAndAttribute = coreDictRepository.findByObjectAndAttribute("MdmHcp", "departName");
                    Map<String, String> collect = bydObjectAndAttribute.stream()
                            .collect(Collectors.toMap(CoreDict::getCode, CoreDict::getLabel));

                    // 放入Step ExecutionContext
                    chunkContext.getStepContext().getStepExecution()
                            .getExecutionContext().put("coreDictMap", collect);

                    return RepeatStatus.FINISHED;
                })
                .listener(promotionListener())
                .build();
    }

    public StepExecutionListener promotionListener() {
        ExecutionContextPromotionListener listener = new ExecutionContextPromotionListener();
        listener.setKeys(new String[]{"coreDictMap"});
        return listener;
    }

    @Bean
    public Step syncHcpStep(ItemProcessor<MdmHcp, Hcp> hcpSyncProcessor) {
        return stepBuilderFactory.get("syncHcpStep")
                .<MdmHcp, Hcp>chunk(CHUNK_SIZE)
                .reader(mceHcpReader())
                .processor(hcpSyncProcessor)
                .writer(hcpJpaItemWriter)
                .faultTolerant()
                .skipLimit(1000)
                .skip(DataIntegrityViolationException.class)
                .skip(ConstraintViolationException.class)
                .skip(IncorrectResultSizeDataAccessException.class)
                .skip(NonUniqueResultException.class)
                .transactionManager(transactionManagerIevent)
                .build();
    }

    public Step syncSpeakerStep() {
        return stepBuilderFactory.get("syncSpeakerStep")
                .<MdmSpeaker, MtSpeakerSchema>chunk(CHUNK_SIZE)
                .reader(mceSpeakerReader())
                .processor(speakerSyncProcessor())
                .writer(mtSpeakerSchemaWriter)
                .faultTolerant()
                .skipLimit(1000)
                .skip(DataIntegrityViolationException.class)
                .skip(ConstraintViolationException.class)
                .skip(IncorrectResultSizeDataAccessException.class)
                .skip(NonUniqueResultException.class)
                .transactionManager(transactionManagerIevent)
                .build();
    }


    public ItemReader<MdmHcp> mceHcpReader() {
        RepositoryItemReader<MdmHcp> reader = new RepositoryItemReader<>();
        reader.setRepository(mdmHcpRepository);
        reader.setMethodName("findAll");
        reader.setSort(Collections.singletonMap("id", Sort.Direction.ASC));
        reader.setPageSize(CHUNK_SIZE);
        return reader;
    }

    public ItemReader<MdmSpeaker> mceSpeakerReader() {
        RepositoryItemReader<MdmSpeaker> reader = new RepositoryItemReader<>();
        reader.setRepository(mdmSpeakerRepository);
        reader.setMethodName("findAll");
        reader.setSort(Collections.singletonMap("id", Sort.Direction.ASC));
        reader.setPageSize(CHUNK_SIZE);
        return reader;
    }


    @StepScope
    @Bean
    public ItemProcessor<MdmHcp, Hcp> hcpSyncProcessor(@Value("#{jobExecutionContext['coreDictMap']}") Map<String, String> coreDictMap) {
        return mdmHcp -> {
            Date date = new Date();
            Hcp hcp = hcpRepository.getHcpByAccCode(mdmHcp.getHcpCode());
            if (hcp == null) {
                hcp = new Hcp();
                hcp.setAccCode(mdmHcp.getHcpCode());
                hcp.setCreateTime(date);
            }
            hcp.setExternalId(mdmHcp.getId());
//            hcp.setIsActive(NumberUtil.parseInt(mdmHcp.getIsActive()));
            hcp.setIsActive("yes".equals(mdmHcp.getHcpStatus())?1:0);
            hcp.setAccName(mdmHcp.getHcpName());
            hcp.setName(mdmHcp.getHcpName());
            hcp.setHcoCode(mdmHcp.getHcoCode());
            hcp.setAccType("doctor");
            hcp.setHcoName(mdmHcp.getHcoName());
            hcp.setIsSpeaker(NumberUtil.parseInt( mdmHcp.getHcpIsSpeaker()));
            if (mdmHcp.getDepartName() != null) {
                hcp.setAccDepart(coreDictMap.getOrDefault(mdmHcp.getDepartName(), mdmHcp.getDepartName()));
            }
            hcp.setHospitalName(mdmHcp.getHcoName());
            hcp.setUpdateTime(date);
            return hcp;
        };
    }

    public ItemProcessor<MdmSpeaker, MtSpeakerSchema> speakerSyncProcessor() {
        return speaker -> {
            Date date = new Date();
            MtSpeakerSchema mtSpeakerSchema = mtSpeakerSchemaRepository.findByAccId(speaker.getHcpId());
            if (mtSpeakerSchema == null) {
                mtSpeakerSchema = new MtSpeakerSchema();
                mtSpeakerSchema.setCreateTime( date);
            }
            mtSpeakerSchema.setAccId(speaker.getHcpId());
            MdmHcp mdmHcp = speaker.getMdmHcp();
            if (mdmHcp != null) {
                mtSpeakerSchema.setHcpId(mdmHcp.getHcpCode());
            }
            mtSpeakerSchema.setActive(1);
            mtSpeakerSchema.setLockOut(0);
            mtSpeakerSchema.setGradeId(covertMceLevelToIevent(speaker.getPaidTiers()));
            mtSpeakerSchema.setTypeId("1484096675048390657");
            mtSpeakerSchema.setUpdateTime(date);
            return mtSpeakerSchema;
        };
    }


    String covertMceLevelToIevent(String paidTiers){
        if (paidTiers == null || paidTiers.isEmpty()) {
            return "1501500861908037634";
        }
        switch (paidTiers){
            case "1": return "1501500861908037634";
            case "2": return "1501500861857705986";
            case "3": return "1508706862223708111";
            default:  return "1508706862223708111";
        }
    }
}